package org.example.flink.cogroup;

import java.util.Date;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.example.flink.source.S1;
import org.example.flink.source.S2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple3;

/**
 * coGroup类似join算子，join是交集； coGroup是全集（关联不上的数据也保留）
 */
public class FCogroup {

    private static Logger logger = LoggerFactory.getLogger(FCogroup.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Tuple3<String, Integer, Integer>> s1 = env.addSource(new S1());
        DataStreamSource<Tuple3<String, Integer, Integer>> s2 = env.addSource(new S2());

        DataStream<String> apply = s1.coGroup(s2).where(new KeySelector<Tuple3<String, Integer, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer, Integer> v1) throws Exception {
                return v1._1();
            }
        }).equalTo(new KeySelector<Tuple3<String, Integer, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer, Integer> v2) throws Exception {
                return v2._1();
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
            .apply(new CoGroupFunction<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>, String>() {
                @Override
                public void coGroup(Iterable<Tuple3<String, Integer, Integer>> iterable,
                    Iterable<Tuple3<String, Integer, Integer>> iterable1, Collector<String> collector)
                    throws Exception {

                    iterable.forEach(v1 -> {
                        String s = new Date() + " " + v1._1() + " (" + v1._2() + "," + v1._3() + ")";
                        logger.info(s);
                        collector.collect(s);

                    });

                    iterable1.forEach(v2 -> {
                        String s = new Date() + " " + v2._1() + " (" + v2._2() + "," + v2._3() + ")";
                        logger.info(s);
                        collector.collect(s);
                    });
                }
            });

        apply.print();

        env.execute("FCogroup");

    }

}
